package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

object Demo05MapPartitions {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo05MapPartitions")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    //    val idRDD: RDD[String] = sc.parallelize(List[String]("1500100001", "1500100002", "1500100003"))
    val idRDD: RDD[String] = sc.textFile("Spark/data/students.txt").map(line => line.split(",")(0))


    // 算子外部的代码在Driver端执行，算子内部的代码是在Task中执行
    // Connection对象以及PreparedStatement对象对未实现序列化接口，所以不能在算子外部定义然后再算子内部使用

    idRDD.map(id => {

      // 直接使用map方法会对数据进行遍历
      // 如果将创建连接的过程放入到map算子内部，那么每一条数据都会创建依次连接
      // 创建、销毁链接是非常耗时的 --> 效率太低
      // 通过id 从MySQL中的student库下的student_like表中获取学生的信息
      // 创建MySQL连接
      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
      println("创建了一次连接")
      val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz from student_like where id=?")
      pSt.setString(1, id)
      val rs: ResultSet = pSt.executeQuery()
      var resStr: String = ""
      while (rs.next()) {
        val id: String = rs.getString("id")
        val name: String = rs.getString("name")
        val age: String = rs.getString("age")
        val gender: String = rs.getString("gender")
        val clazz: String = rs.getString("clazz")
        resStr = s"$id,$name,$age,$gender,$clazz"
      }
      // 销毁连接
      pSt.close()
      conn.close()
      resStr
    })
      .foreach(println)

    // 通常需要与外部建立连接时并从外部系统获取数据时 则最好使用mapPartitions算子进行处理
    val resRDD: RDD[String] = idRDD
      .mapPartitions(iter => {
        /**
         * mapPartitions 转换算子
         * 该算子会对RDD中的每一个分区依次处理
         * 分区的数量会远小于数据的条数
         * 与map算子相比 可以大大减少连接创建以及销毁的次数
         * 使用mapPartitions能够让每个Task共用一个外部链接
         *
         */
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")
        println("创建了1次连接")

        val pSt: PreparedStatement = conn.prepareStatement("select id,name,age,gender,clazz from student_like where id=?")

        // mapPartitions会将每个分区的数据变成迭代器，这里相当于对每个分区的数据进行处理
        val resIter: Iterator[String] = iter.map(id => {
          pSt.setString(1, id)
          val rs: ResultSet = pSt.executeQuery()
          var resStr: String = ""
          while (rs.next()) {
            val id: String = rs.getString("id")
            val name: String = rs.getString("name")
            val age: String = rs.getString("age")
            val gender: String = rs.getString("gender")
            val clazz: String = rs.getString("clazz")
            resStr = s"$id,$name,$age,$gender,$clazz"
          }
          resStr
        })
        // 销毁连接 再mapPartitions中不能销毁连接，等程序结束会自动销毁连接
        //        pSt.close()
        //        conn.close()

        resIter
      })

    resRDD.foreach(println)

  }

}
